package com.huilian.iotbox.mqtt.server.emqx.consumer;

import com.alibaba.fastjson.JSON;
import com.huilian.iotbox.aliyun.service.PubService;
import com.huilian.iotbox.data.enums.MessageLogResultEnum;
import com.huilian.iotbox.data.po.MessageLog;
import com.huilian.iotbox.data.producer.MessageLogRabbit;
import com.huilian.iotbox.data.service.ErrorLogCommonService;
import com.huilian.iotbox.mqtt.server.emqx.subscribe.AmqpClient;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.Map;
import java.util.Random;

@Slf4j
@Component
public class RabbitEmqxReceiver {

    @Value("${huilian.iot.productKey}")
    private String productKey;
    @Autowired
    private PubService pubService;
    @Autowired
    private ErrorLogCommonService errorLogCommonService;
    @Autowired
    private MessageLogRabbit messageLogRabbit;
    private static Random emqxConnectionRandom = new Random();

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "${spring.rabbitmq.listener.iotBoxEmqxPubMessage.queue.name}",
            durable = "${spring.rabbitmq.listener.iotBoxEmqxPubMessage.queue.durable}"),
            exchange = @Exchange(value = "${spring.rabbitmq.listener.iotBoxEmqxPubMessage.exchange.name}",
                    durable = "${spring.rabbitmq.listener.iotBoxEmqxPubMessage.exchange.durable}",
                    type = "${spring.rabbitmq.listener.iotBoxEmqxPubMessage.exchange.type}"),
            ignoreDeclarationExceptions = "${spring.rabbitmq.listener.iotBoxEmqxPubMessage.exchange.ignoreDeclarationExceptions}",
            key = "${spring.rabbitmq.listener.iotBoxEmqxPubMessage.key}"))
    @RabbitHandler
    public void emqxPubMessage(@Payload String json, Channel channel, @Headers Map<String, Object> headers) throws IOException {
        log.info("EMQX send:" + json);
        MessageLog messageLog = JSON.parseObject(json, MessageLog.class);

        if (StringUtils.isEmpty(messageLog.getContent())) {
            errorLogCommonService.writerInfoErrorLog("EMQ发送内容为空", messageLog);
            return;
        }
        try {
            MqttMessage message = new MqttMessage(messageLog.getContent().getBytes());
            message.setQos(AmqpClient.qos);

            final int emqIndex = this.getIndex();
            MqttClient mqttClient = AmqpClient.mqttClientList.get(emqIndex);
            if (!mqttClient.isConnected()) {
                log.error("emqIndex:" + emqIndex);
                AmqpClient.mqttClientList.remove(emqIndex);
                throw new RuntimeException(mqttClient.getClientId() + "未连接，重新获取MqttClient发送消息");
            }

            mqttClient.publish(messageLog.getTopic(), message);
        } catch (Exception e) {
            e.printStackTrace();
            messageLog.setResult(MessageLogResultEnum.FAIL.getCode());
            log.error("EMQX发送消息不成功：{}", e.getMessage());
            pubService.pub(messageLog.getContent(), messageLog.getDeviceNo(), productKey);
            errorLogCommonService.writerInfoException("EMQX发送消息不成功", messageLog, e);
        } finally {
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            //手工ACK
            channel.basicAck(deliveryTag, true);
            messageLog.setResult(MessageLogResultEnum.SUCCESS.getCode());
            if (messageLog.getContent().length() >= 8 && messageLog.getContent().substring(2, 8).equalsIgnoreCase("160001")) {
                return;
            }
            messageLogRabbit.sendMessageLog(messageLog);
        }
    }

    private int getIndex() {
        return RabbitEmqxReceiver.emqxConnectionRandom.nextInt(AmqpClient.mqttClientList.size() - 1);
    }
}
